package com.chb.userbehavioranalysis.order

import org.apache.flink.api.common.state.{ValueState, ValueStateDescriptor}
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.KeyedProcessFunction
import org.apache.flink.streaming.api.scala._
import org.apache.flink.util.Collector

/**
 * 使用定时器和State检测超时事件
 */
object OrderTimeoutWithoutCep {

    val orderTimeoutOutputTag = new OutputTag[OrderResult]("orderTimeout")

    def main(args: Array[String]): Unit = {
        val env = StreamExecutionEnvironment.getExecutionEnvironment
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
        env.setParallelism(1)

        // 读取订单数据
        val resource = getClass.getResource("/OrderLog.csv")
//        val orderEventStream = env.readTextFile(resource.getPath)
        val orderEventStream = env.socketTextStream("10.0.0.201", 8888)
            .map(data => {
                val dataArray = data.split(",")
                OrderEvent(dataArray(0).trim.toLong, dataArray(1).trim, dataArray(2).trim, dataArray(3).trim.toLong)
            })
            .assignAscendingTimestamps(_.eventTime * 1000L)
            .keyBy(_.orderId)

        // 定义process function进行超时检测
//        val orderResultStream = orderEventStream.process(new OrderTimeOutWaring())

        val orderResultStream = orderEventStream.process(new OrderTimeOutMatch())

        orderResultStream.print()
        orderResultStream.getSideOutput(orderTimeoutOutputTag).print("timeOut")

        env.execute("order timeout without cep job")
    }

    class OrderTimeOutMatch() extends KeyedProcessFunction[Long, OrderEvent, OrderResult] {
        // 保存pay是否检测到的状态
        lazy val isPayedState: ValueState[Boolean] = getRuntimeContext.getState(new ValueStateDescriptor[Boolean]("pay", classOf[Boolean]))
        // 保存定时器时间
        lazy val timeState: ValueState[Long] = getRuntimeContext.getState(new ValueStateDescriptor[Long]("timeState", classOf[Long]))

        override def processElement(i: OrderEvent, context: KeyedProcessFunction[Long, OrderEvent, OrderResult]#Context, collector: Collector[OrderResult]): Unit = {
            val isPayed = isPayedState.value()
            val timeTs = timeState.value()
            // pay先到， 等待create
            if (i.orderId == 34768) {
                println(isPayed, i.eventType)
            }

            // 根据事件状态做不同处理
            if (i.eventType == "create" ) {
                // 1.1、pay 已经来了，
                if (isPayed) {
                    // 时间乱序
                    context.output(orderTimeoutOutputTag, OrderResult(i.orderId, "pay success"))
                    context.timerService().deleteEventTimeTimer(timeTs)
                    isPayedState.clear()
                    timeState.clear()
                } else{
                    // pay没有来， 注册定时器等待
                    val ts = i.eventTime * 1000L + 15*60*1000L
                    context.timerService().registerEventTimeTimer(ts)
                    timeState.update(ts)
                }
            } else if (i.eventType == "pay"){
                // 通过timeTs判断create是否已经来了
                if (timeTs > 0) {
                    // 定时器没有超时
                    if (timeTs > i.eventTime*1000L) {
                        // 匹配成功
                        collector.collect(OrderResult(i.orderId, "pay success..."))
                    } else {
                        // 超时, 放到侧输出流
                        context.output(orderTimeoutOutputTag, OrderResult(i.orderId, "payed but order timeout"))
                    }
                    context.timerService().deleteEventTimeTimer(timeTs)
                    isPayedState.clear()
                    timeState.clear()
                } else {
                    // pay先到， create 还未到
                    isPayedState.equals(true)
                    println(i.eventTime * 1000L)
                    context.timerService().registerEventTimeTimer(i.eventTime * 1000L)
                    timeState.update(i.eventTime*1000L)
                }
            }
        }

        override def onTimer(timestamp: Long, ctx: KeyedProcessFunction[Long, OrderEvent, OrderResult]#OnTimerContext, out: Collector[OrderResult]): Unit = {
            val isPayed = isPayedState.value()
            if (isPayed) {
                ctx.output(orderTimeoutOutputTag, OrderResult(ctx.getCurrentKey, "Order pay  but not log create"))
            } else {
                ctx.output(orderTimeoutOutputTag, OrderResult(ctx.getCurrentKey, "Order TimeOut...."))
            }
            // 状态清空
            isPayedState.clear()
            timeState.clear()
        }
    }

    class OrderTimeOutWaring() extends KeyedProcessFunction[Long, OrderEvent, OrderResult] {
        // 保存pay是否检测到的状态
        lazy val isPayedState: ValueState[Boolean] = getRuntimeContext.getState(new ValueStateDescriptor[Boolean]("pay", classOf[Boolean]))

        override def processElement(i: OrderEvent, context: KeyedProcessFunction[Long, OrderEvent, OrderResult]#Context, collector: Collector[OrderResult]): Unit = {
            val isPayed = isPayedState.value()

            if (i.eventType == "create" && !isPayed) {//订单创建，注册一个定时器L
                context.timerService().registerEventTimeTimer(i.eventTime * 1000L + 15*60*1000) // 设置定时器等待时间
            } else if (i.eventType == "pay"){
                isPayedState.update(true)
            }
        }

        override def onTimer(timestamp: Long, ctx: KeyedProcessFunction[Long, OrderEvent, OrderResult]#OnTimerContext, out: Collector[OrderResult]): Unit = {
            val isPayed = isPayedState.value()
            if (isPayed) {
                out.collect(OrderResult(ctx.getCurrentKey, "Order pay success...."))
            } else {
                out.collect(OrderResult(ctx.getCurrentKey, "Order TimeOut...."))
            }
            // 状态清空
            isPayedState.clear()
        }
    }
}



